package cn.gxufe.reliable.message.admin.utils;

import cn.gxufe.reliable.message.admin.domain.RMessage;
import cn.gxufe.reliable.message.admin.service.impl.MessageEventHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author yingjie.rong
 **/
public class PublishUtils {

    private  static PublishUtils publishUtils;

    private static final Logger logger = LoggerFactory.getLogger(PublishUtils.class);
    private static final int POOL_SIZE = Runtime.getRuntime().availableProcessors();
    private static final long KEEP_ALIVE_TIME = 0L;
    private static final long SLEEP_TIME = 100;
    private static MessageEventHandler messageEventHandler;
    private static ThreadPoolExecutor threadPoolExecutor;
    private static LinkedBlockingQueue<RMessage> queue;

    private PublishUtils(){
        queue = new LinkedBlockingQueue<>();
    }

    public synchronized static PublishUtils getPublishUtils(){
        if( publishUtils == null ){
            publishUtils = new PublishUtils();
        }
        return publishUtils;
    }

    @Autowired
    public void setMessageEventHandler(MessageEventHandler messageEventHandler) {
        PublishUtils.messageEventHandler = messageEventHandler;
    }

    public static void publish(RMessage message){
        if( queue == null || messageEventHandler == null ){
            logger.debug("queue is not open !");
            return;
        }
        logger.debug("add to   queue , id = {} ,status = {}" , message.getId(),message.getStatus());
        try {
            queue.put(message);
        } catch (InterruptedException e) {
            logger.warn("publish",e);
        }
    }

    @PostConstruct
    private void post(){
        logger.info("submit  processRMessage ");
        PublishUtils.threadPoolExecutor =  new ThreadPoolExecutor(POOL_SIZE, POOL_SIZE,KEEP_ALIVE_TIME, TimeUnit.MILLISECONDS,new LinkedBlockingQueue<>());
        threadPoolExecutor.submit( () -> {
            for (;;){
                processRMessage(queue.take());
            }
        }  );
    }

    private void processRMessage(RMessage rMessage){
        logger.info("processRMessage id = {}, key = {}, status = {} ",rMessage.getId(),rMessage.getKey(),rMessage.getStatus());
        try {
            Thread.sleep(SLEEP_TIME);
            messageEventHandler.onEvent(rMessage);
        } catch (InterruptedException e) {
            logger.error("process RMessage error !",e);
        }
    }
}
